privateGPT MCP Server

import socket import ssl import json import argparse import sys sys.stdout.reconfigure(encoding='utf-8') sys.stdin.reconfigure(encoding='utf-8') def send_mcp_request(server_ip, server_port, token, question, use_public, groups=None, language="de", use_ssl=True, accept_self_signed=False): """ Sends a question to an MCP server and retrieves the response. :param server_ip: IP address of the MCP server :param server_port: Port number of the MCP server :param token: Authentication token for the MCP server :param question: The question to send :param use_public: Whether to use the public knowledge base :param groups: List of groups for retrieval-augmented generation :param language: Language code for the request :param use_ssl: Whether to use SSL/TLS for the connection :param accept_self_signed: Whether to accept self-signed certificates :return: Response from the server """ # Prepare the request payload payload = { "command": "chat", "token": token, "arguments": { "question": question, "usePublic": use_public, "groups": groups or [], "language": language } } # Convert the payload to a JSON string payload_json = json.dumps(payload) # Initialize socket variables raw_socket = None client_socket = None try: # Create a socket object raw_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) raw_socket.settimeout(10) # Establish SSL/TLS connection if required if use_ssl: context = ssl.create_default_context() if accept_self_signed: context.check_hostname = False context.verify_mode = ssl.CERT_NONE client_socket = context.wrap_socket(raw_socket, server_hostname=server_ip) else: client_socket = raw_socket # Connect to the server client_socket.connect((server_ip, server_port)) # Send the request client_socket.sendall(payload_json.encode('utf-8')) # Receive the response response = b"" while True: part = client_socket.recv(4096) if not part: break response += part # Decode the response return json.loads(response.decode('utf-8')) except ssl.SSLError: return {"status": "error", "message": "Connection failed: Server and/or Client may require TLS encryption. Please enable SSL/TLS."} except Exception as e: return {"status": "error", "message": str(e)} finally: if client_socket is not None: try: client_socket.shutdown(socket.SHUT_RDWR) except: pass client_socket.close() if __name__ == "__main__": parser = argparse.ArgumentParser(description="Send a question to the MCP server and retrieve the response.") parser.add_argument("--server-ip", required=True, help="IP address of the MCP server") parser.add_argument("--server-port", required=True, type=int, help="Port number of the MCP server") parser.add_argument("--token", required=True, help="Authentication token") parser.add_argument("--question", required=True, help="The question to ask the MCP server") parser.add_argument("--use-public", action="store_true", help="Use the public knowledge base") parser.add_argument("--groups", nargs="*", help="List of groups for retrieval-augmented generation", default=[]) parser.add_argument("--language", default="de", help="Language code for the request (default: 'de')") parser.add_argument("--use-ssl", action="store_true", help="Connect using SSL/TLS") parser.add_argument("--accept-self-signed", action="store_true", help="Accept self-signed certificates (disable certificate verification)") args = parser.parse_args() # Send the question to the MCP server response = send_mcp_request( server_ip=args.server_ip, server_port=args.server_port, token=args.token, question=args.question, use_public=args.use_public, groups=args.groups, language=args.language, use_ssl=args.use_ssl, accept_self_signed=args.accept_self_signed ) print("Response from server:", json.dumps(response, indent=2, ensure_ascii=False))